/*
 * @(#)CreateExecSqlChannel.java   1.0  2021年11月8日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.util;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.swift.core.spring.Spring;
import com.swift.dts.loadconfig.FileTablesConfig;
import com.swift.dts.loadconfig.model.TablesConfig;
import com.swift.dts.model.DtsBinlogMessage;
import com.swift.dts.model.SqlObject;
import com.swift.util.date.DateUtil;
import com.swift.util.security.MD5Utils;
import com.swift.util.type.TypeUtil;

/**
 * 把event处理成可执行的SQL事件
 * 
 * @author zhengjiajin
 * @version 1.0 2021年11月8日
 */
public class CreateExecSql {

    public static void setExecSql(DtsBinlogMessage message) {
        if (message == null) return;
        if (message.getIsDdl()) {
            setDdlSql(message);
            return;
        }
        if ("INSERT".equals(message.getType())) {
            message.setExecSqlList(insertSql(message));
            ;
        }
        if ("UPDATE".equals(message.getType())) {
            message.setExecSqlList(updateSql(message));
        }
        if ("DELETE".equals(message.getType())) {
            message.setExecSqlList(deleteSql(message));
        }
    }

    private static void setDdlSql(DtsBinlogMessage message) {
        if (message == null) return;
        if (!message.getIsDdl()) return;
        if (TypeUtil.isNull(message.getSql())) return;
        if (TypeUtil.isNull(message.getTable())) return;
        String sql = message.getSql();
        String table = message.getTable();
        int tableSite = sql.toLowerCase().indexOf(table.toLowerCase());
        String sqlStart = sql.substring(0, tableSite);
        String sqlEnd = sql.substring(tableSite + table.length());
        String execSql = sqlStart + dbTable(message.getDatabase(), message.getTable()) + sqlEnd;
        message.setExecSqlList(Arrays.asList(new SqlObject(null, unionKey(execSql), execSql)));
    }

    public static String unionKey(String execSql) {
        return MD5Utils.getMD5Code(cleanSql(execSql));
    }
    //去除形响匹配因素的键值
    public static String cleanSql(String execSql) {
        // 去掉日期格式的字符串，防跨区问题yyyy-MM-dd HH:mm:ss
        String regexDate = "'[0-9]{4}[-][0-9]{2}[-][0-9]{2}[ ][0-9]{2}[:][0-9]{2}[:][0-9]{2}'";
        execSql = replaceRegex(execSql, regexDate);
        String regexTimestamp = "FROM_UNIXTIME\\([0-9]{0,13}\\)";
        execSql = replaceRegex(execSql, regexTimestamp);
        //去掉SQL \的问题
        execSql = execSql.replace("\\", "");
        execSql = execSql.replace("'", "");
        return execSql;
    }

    private static String replaceRegex(String execSql, String regex) {
        Pattern pattern = Pattern.compile(regex);
        Matcher matcher = pattern.matcher(execSql);
        while (matcher.find()) {
            execSql = execSql.replace(matcher.group(), "");
        }
        return execSql;
    }

    private static String dbTable(DtsBinlogMessage message) {
        return dbTable(message.getDatabase(), message.getTable());
    }

    public static String dbTable(String dbName, String tableName) {
        FileTablesConfig tableConfig = Spring.getBean(FileTablesConfig.class);
        TablesConfig tableNameChange = tableConfig.findConfig("tableNameChange");
        if (TypeUtil.isNull(tableNameChange)) return tableName;
        String change = TypeUtil.toString(tableNameChange.getRegex());
        if (TypeUtil.isNull(change)) return tableName;
        change = change.replace("${db_name}", dbName);
        change = change.replace("${db_table_name}", tableName);
        return change;
    }
    
    private static String doColumn(String column) {
        return column.toLowerCase();
    }

    /**
     * 更新语句
     */
    private static List<SqlObject> updateSql(DtsBinlogMessage message) {
        if (TypeUtil.isNull(message.getData())) return null;
        List<SqlObject> list = new ArrayList<SqlObject>(message.getData().size());
        for (Map<String, Object> data : message.getData()) {
            if (TypeUtil.isNull(data)) continue;
            List<String> columnList = sortColumn(data.keySet());
            List<String> pkList = sortColumn(message.getPkNames());
            StringBuffer sql = new StringBuffer("update " + dbTable(message).toLowerCase() + " set ");
            for (String column : columnList) {
                if (isPk(message.getPkNames(), column)) continue;
                if(isCleanColumn(column)) continue;
                sql.append(doColumn(column) + "=" + value(data.get(column)));
                sql.append(",");
            }
            sql.delete(sql.length() - 1, sql.length());
            sql.append(" where ");
            int i = 0;
            for (String pk : pkList) {
                if (i > 0) {
                    sql.append(" and ");
                }
                sql.append(doColumn(pk) + whereIsNull(value(data.get(pk))));
                i++;
            }
            doingUpdateTime(sql, data);

            String pkKey = getPkKey(pkList, data);
            list.add(new SqlObject(pkKey, unionKey(sql.toString()), sql.toString()));
        }
        return list;
    }

    private static String[] updateTimeColumn = { "update_time", "updatetime" };

    private static void doingUpdateTime(StringBuffer sql, Map<String, Object> data) {
        if(TypeUtil.isNull(data)) return;
        for (String column : updateTimeColumn) {
            for(String dataColumn:data.keySet()) {
                if(!column.equalsIgnoreCase(dataColumn)) continue;
                // 对有update_time,且为时间挫的字段进行特别处理,对大于时间的则不更新
                Object update_time = data.get(dataColumn);
                if (TypeUtil.isNotNull(update_time) && update_time instanceof Timestamp) {
                    sql.append(" and " + doColumn(dataColumn) + "<=").append(value(update_time));
                }
            }
        }
    }

    private static String whereIsNull(String value) {
        if ("null".equalsIgnoreCase(value)) return " is null ";
        return " = " + value;
    }

    public static String value(Object value) {
        if (value == null) return "null";
        if (value instanceof Timestamp) {
            long time = ((Timestamp) value).getTime()/1000;
            if(time<0) return "'" + DateUtil.formatDate((Date) value) + "'";
            if(time>2147483647) return "'" + DateUtil.formatDate((Date) value) + "'";
            return "FROM_UNIXTIME(" +  time + ")";
        }
        if (value instanceof Date) return "'" + DateUtil.formatDate((Date) value) + "'";
        if (TypeUtil.isNumber(value)) return TypeUtil.toString(value);
        String value_str = TypeUtil.toString(value);
        if (value_str.equalsIgnoreCase("null")) return "null";
        if ("true".equalsIgnoreCase(value_str)) return "1";
        if ("false".equalsIgnoreCase(value_str)) return "0";
        value_str = formatValue(value_str);
        return "'" + value_str + "'";
    }

    private static boolean isPk(List<String> pkNames, String column) {
        if (TypeUtil.isNull(pkNames)) return false;
        return pkNames.contains(column);
    }

    /**
     * 删除语句
     */
    private static List<SqlObject> deleteSql(DtsBinlogMessage message) {
        if (TypeUtil.isNull(message.getData())) return null;
        List<SqlObject> list = new ArrayList<SqlObject>(message.getData().size());
        for (Map<String, Object> data : message.getData()) {
            if (TypeUtil.isNull(data)) continue;
            List<String> pkList = sortColumn(message.getPkNames());
            StringBuffer sql = new StringBuffer("delete from " + dbTable(message).toLowerCase());
            sql.append(" where ");
            int i = 0;
            for (String pk : pkList) {
                if (i > 0) {
                    sql.append(" and ");
                }
                sql.append(doColumn(pk) + whereIsNull(value(data.get(pk))));
                i++;
            }
            String pkKey = getPkKey(pkList, data);
            list.add(new SqlObject(pkKey, unionKey(sql.toString()), sql.toString()));
        }
        return list;
    }

    /**
     * 插入语句
     */
    private static List<SqlObject> insertSql(DtsBinlogMessage message) {
        if (TypeUtil.isNull(message.getData())) return null;
        List<SqlObject> list = new ArrayList<SqlObject>(message.getData().size());
        for (Map<String, Object> data : message.getData()) {
            if (TypeUtil.isNull(data)) continue;
            List<String> columnList = sortColumn(data.keySet());
            List<String> pkList = sortColumn(message.getPkNames());
            StringBuffer sql = new StringBuffer("insert into " + dbTable(message).toLowerCase() + " (");
            for (String column : columnList) {
                if(isCleanColumn(column)) continue;
                sql.append(doColumn(column) + ",");
            }
            sql.delete(sql.length() - 1, sql.length());
            sql.append(") VALUES (");
            for (String column : columnList) {
                if(isCleanColumn(column)) continue;
                sql.append(value(data.get(column)) + ",");
            }
            sql.delete(sql.length() - 1, sql.length());
            sql.append(")");

            String pkKey = getPkKey(pkList, data);
            list.add(new SqlObject(pkKey, unionKey(sql.toString()), sql.toString()));
        }
        return list;
    }
    
    private static final String[] cleanColumn = {"__#alibaba_rds_row_id#__"};
    
    private static boolean isCleanColumn(String column) {
        for(String clean:cleanColumn) {
            if(clean.equalsIgnoreCase(column)) return true;
        }
        return false;
    }
    
    public static String formatValue(String value) {
        if (TypeUtil.isNull(value)) return value;
        value = value.replace("'", "''");
        value = value.replace("\\", "\\\\");
        return value;
    }

    private static String getPkKey(List<String> pks, Map<String, Object> data) {
        StringBuffer sb = new StringBuffer();
        if (TypeUtil.isNull(pks)) { // 所有键同为主键
            for (Object value : data.values()) {
                sb.append(value).append(";");
            }
            return sb.toString();
        }
        for (String pk : pks) {
            sb.append(data.get(pk)).append(";");
            return sb.toString();
        }
        return "";
    }
    
    public static String getPkKey(CommonMessage message) {
        if (TypeUtil.isNull(message.getData())) return "";
        for (Map<String, Object> data : message.getData()) {
            if (TypeUtil.isNull(data)) continue;
            List<String> pkList = sortColumn(message.getPkNames());
            String pkKey = getPkKey(pkList, data);
            return pkKey;
        }
        return "";
    }
    
    private static List<String> sortColumn(Collection<String> column){
        if(TypeUtil.isNull(column)) return new ArrayList<>();
        List<String> list = new ArrayList<>(column);
        Collections.sort(list);
        return list;
    }

}
